package com.chencong.online.driver.dwd;

import com.chencong.online.bean.OrderEventBean;
import com.chencong.online.bean.OrderResultBean;
import com.chencong.online.env.FlinkEnv;
import com.chencong.online.function.OrderEventMapFunc;
import com.chencong.online.function.OrderPayDetectProcessFunc;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.OutputTag;

import java.net.URL;

/**
 * @Author chencong
 * @Description利用状态检测订单
 * @Date 9:30 下午 2022/1/5
 * @Param
 **/
public class DwdOrderPayWithStateDriver {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = FlinkEnv.FlinkDataStreamRunEnv();
        env.setParallelism(1);
        //1、source
        URL resource = DwdOrderPayTimeOutDriver.class.getResource("/OrderLog.csv");
        DataStreamSource<String> inputDS = env.readTextFile(resource.getPath());
//        inputDS.print("初始化数据");
        //2、transform
        //将数据格式化并提取时间戳
        SingleOutputStreamOperator<OrderEventBean> orderEventBeanDS = inputDS.map(new OrderEventMapFunc())
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<OrderEventBean>forMonotonousTimestamps() //升序
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderEventBean>() {
                            @Override
                            public long extractTimestamp(OrderEventBean element, long recordTimestamp) {
                                return element.getTimestamp() * 1000L;
                            }
                        })
                );

        //利用状态检测，正常匹配流在主流，超时流在侧输出流
        SingleOutputStreamOperator<OrderResultBean> orderPayProcessDS = orderEventBeanDS
                .keyBy(OrderEventBean::getOrderId)
                .process(new OrderPayDetectProcessFunc());

        orderPayProcessDS.print("主流！！");

        orderPayProcessDS.getSideOutput(new OutputTag<OrderResultBean>("order-timeout") {
        }).print("侧输出流！！");
        env.execute("DwdOrderPayWithStateDriver");
    }
}
